package com.compomics.proteocloud.sqs;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import com.compomics.proteocloud.util.AWSUtils;
import com.xerox.amazonws.sqs2.Message;
import com.xerox.amazonws.sqs2.MessageQueue;
import com.xerox.amazonws.sqs2.SQSException;
import com.xerox.amazonws.sqs2.SQSUtils;

/**
 * The MessageManager handles the sending, retrieving and deleting of the messages from the 
 * DataQueue. Note that messages are retrieved in arbitrary fashion as SQS provides this feature.
 * 
 * @author Thilo Muth
 *
 */
public class MessageManager {
	
	/**
	 * The Properties object.
	 */
	private Properties props;
	
	/**
	 * The Messagequeue object.
	 */
	private MessageQueue queue;
	
	/**
	 * Constructor of the MessageManager.
	 * @param props
	 * @param queuename
	 */
	public MessageManager(Properties props, String queuename) throws SQSException {
		this.props = props;		
		initQueue(queuename);
	}
	
	/**
	 * Initializes the Queue.
	 */
	private void initQueue(String queuename) throws SQSException {
		queue = SQSUtils.connectToQueue(queuename, props.getProperty(AWSUtils.AWS_ACCESS_KEY_PROPERTY_NAME), props.getProperty(AWSUtils.AWS_SECRET_KEY_PROPERTY_NAME));
	}
	
	/**
	 * Receive an arbitrary message from the SQS, using default visibility timeout.
	 * @throws SQSException 
	 */
	public Message receiveMessage() throws SQSException {	
		return queue.receiveMessage();			
	}
	
	/**
	 * Receive an arbitrary message from the SQS, using specified visibility timeout.
	 * @param visibilityTimeout
	 * @throws SQSException 
	 */
	public Message receiveMessage(int visibilityTimeout) throws SQSException {
		return queue.receiveMessage(visibilityTimeout);	
	}
	
	/**
	 * Receive all messages from the SQS.	 * 
	 * @return Message[] The received messages.
	 * @throws SQSException
	 */
	public List<Message> receiveMessages() throws SQSException{
		int number = queue.getApproximateNumberOfMessages();
		Message msg = null;
		List<Message> messageList = null;
		int count = 0;
		//Thread.sleep((long) (60000 / pollRate));
		if(number > 0){
			messageList = new ArrayList<Message>();
			while (true) {
				msg = queue.receiveMessage(1);				
				if(msg != null){
					deleteMessage(msg);
					//messageList.add(msg);					
					count++;
					if(count == number) break;
				}
			}
		} 
		return messageList;
	}
	
	/**
	 * Send a specified message to an SQS queue.	 *
	 * @throws SQSException 
	 */
	public void sendMessage(String filename, String projectTitle, String dbName, String dbUrl, String dbUser, String dbPassword, String fastaDB, String fragmentTol, String precursorTol, String precursorUnit, String bucket) throws SQSException {
        queue.setEncoding(true);
        queue.sendMessage(filename + " " + projectTitle + " " + dbName + " "  + dbUrl + " "  + dbUser + " " + dbPassword + " " + fastaDB + " " + fragmentTol + " " + precursorTol + " " + precursorUnit + " " + bucket);

	}
	
	/**
	 * Delete a specified message from the queue.
	 * @param message
	 * @throws SQSException 
	 */
	public void deleteMessage(Message message) throws SQSException{		
		queue.deleteMessage(message.getReceiptHandle());		
	}
	
	/**
	 * Deletes the message by the specific messageid.
	 * @param messageId
	 * @throws SQSException
	 */
	public void deleteMessageByMessageId(String messageId) throws SQSException{
		List<Message> messageList;
		
		// Try to receive the messages.		
		messageList = this.receiveMessages();
		if(messageList != null){
			// Iterate over all messages.
			for (Message message : messageList) {
				if(message.getMessageId().equalsIgnoreCase(messageId)){
					queue.deleteMessage(message.getReceiptHandle());	
				}
			}
		}
	}
}
